"""
Case Type   : 极致RTO支持备机可读
Case Name   : 资源池化网络复制双集群，备集群开启极致RTO，failover场景下验证备集群可读
Create At   : 2025/04
Owner       : haomeng
Description :
    1、备集群内开启极致RTO特性，重启使其生效
    2、主集群内创建行存段页式range分区表，插入数据
    3、备集群首备节点查看表数据
    4、备集群从备节点查看表数据
    5、stop掉首备节点，查看集群状态
    6、在新的首备节点内查看表数据
    7、在新的从备节点查看表数据
    8、清理环境，恢复环境初始状态
Expect      :
    1、设置参数成功，重启成功
    2、创建成功，插入数据成功
    3、查看成功，与主集群内查看数据一致
    4、查看失败，从备节点不可读
    5、原首备节点down，集群内产生新的首备节点
    6、查看成功
    7、查看失败不可读
    8、清理成功，恢复环境成功
History     :
"""

import os
import time
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class DisasterRTOStandbyRead(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.p_pri_sh = CommonSH('P_PrimaryDbUser')
        self.s_comsh = [CommonSH(n) for n in ('S_PrimaryDbUser', 'S_Standby1DbUser', 'S_Standby2DbUser')]
        self.constant = Constant()

        self.param_list = ['recovery_parse_workers', 'recovery_redo_workers', 'ss_disaster_mode']
        self.value_list = ['4', '4']
        self.default_value = []
        self.table = 'ss_rto_007'
        self.table_new = 'ss_rto_007_new'

        self.log.info('---校验当前环境是否为网络复制双集群---')
        mode = self.p_pri_sh.execut_db_sql(f'show {self.param_list[2]};')
        self.log.info(mode)
        self.assertIn('stream', mode, '当前环境不是网络复制双集群')

    def test_rto_standby_read(self):
        text = '-----step1:集群内开启极致RTO特性，重启使其生效;  expect:设置成功，重启成功-----'
        self.log.info(text)

        for param in self.param_list[:2]:
            value = self.s_pri_sh.execut_db_sql(f'show {param}')
            self.default_value.append(value.splitlines()[-2].strip())
        self.log.info(self.default_value)

        for param in self.param_list[:2]:
            set_param = self.s_pri_sh.execute_gsguc('set',
                                                    self.constant.GSGUC_SUCCESS_MSG,
                                                    f'{param} = {self.value_list[self.param_list.index(param)]}')
            self.log.info(set_param)
            self.assertTrue(set_param, '参数设置失败')

        self.log.info('---重启集群使参数生效---')
        restart = self.s_sta_sh.restart_db_cluster()
        self.log.info(restart)
        self.assertTrue(restart, '集群重启失败')

        self.log.info('---查看参数值修改是否成功---')
        for param in self.param_list[:2]:
            value = self.s_pri_sh.execut_db_sql(f'show {param};')
            self.assertEqual(value.splitlines()[-2].strip(),
                             self.value_list[self.param_list.index(param)], '参数未修改成功')

        text = '-----step2:主集群内创建行存段页式range分区表，插入数据  expect:创建成功，插入数据成功-----'
        self.log.info(text)

        sql1 = f"drop table if exists {self.table};" \
               f"create table {self.table}(id int, name varchar(200), age int, tel int) with(segment=on) " \
               f"partition by range(tel) " \
               f"(partition p1 values less than(5000), partition p2 values less than(10000), " \
               f"partition p3 values less than(15000), partition p4 values less than(20000), " \
               f"partition p5 values less than(25000), partition p6 values less than(30000), " \
               f"partition p7 values less than(40000), partition p8 values less than(50000)); " \
               f"declare    " \
               f"    i integer;" \
               f"begin" \
               f"    for i in 1..100000 loop" \
               f"    insert into {self.table} values(random()*100,'name_'||random()*10,random()*100,random()*10000);" \
               f"    end loop;" \
               f"end;" \
               f"alter table {self.table} rename to {self.table_new};" \
               f"alter table {self.table_new} add new_col1 int, add new_col2 text first, add new_col3 float after new_col2;" \
               f"update {self.table_new} set tel = tel * 2 where id%2 != 0 and id < 1000;" \
               f"delete from {self.table_new} where id < 20;"
        self.log.info(sql1)

        res1 = self.p_pri_sh.execut_db_sql(sql1)
        self.log.info(res1)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS, res1, '建表失败')
        self.assertIn(self.constant.CREATE_ANONYMOUS_BLOCK_SUCCESS_MSG, res1, '匿名块执行失败，插入数据失败')
        self.assertIn(self.constant.ALTER_TABLE_MSG, res1, 'alter表失败')
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG, res1, 'update表数据失败')
        self.assertIn(self.constant.DELETE_SUCCESS_MSG, res1, 'delete表数据失败')

        text = '-----step3:备集群首备节点查看表数据  expect:查看成功-----'
        self.log.info(text)

        self.log.info('---等待备集群数据同步---')
        flag = self.s_comsh[0].check_data_consistency()
        self.assertTrue(flag, '数据同步失败')

        sql2 = f"select count(*) from {self.table_new};"
        self.log.info(sql2)
        select_res = self.s_comsh[0].execut_db_sql(sql2)
        self.log.info(select_res)

        p_select = self.p_pri_sh.execut_db_sql(sql2)
        self.log.info(p_select)
        self.assertEqual(p_select.splitlines()[-2].strip(), select_res.splitlines()[-2].strip(), '查询结果不一致')

        text = '-----step4:备集群从备节点查看表数据  expect:查看失败,从备节点不可读-----'
        self.log.info(text)

        ss_select = self.s_comsh[1].execut_db_sql(sql2)
        self.log.info(ss_select)
        self.assertIn(self.constant.disaster_read_fail_msg, ss_select, '从备节点查询成功')

        text = '-----step5:stop掉首备节点，查看集群状态  expect:原首备节点down，集群内产生新的首备节点-----'
        self.log.info(text)

        stop = self.s_comsh[0].exec_cm_ctl('stop', '-n 1', f'-D {macro.DB_INSTANCE_PATH}')
        self.log.info(stop)
        self.assertIn(self.constant.stop_success_msg, stop, '执行失败' + text)

        self.log.info('---查看集群状态---')
        status = self.s_comsh[0].exec_cm_ctl('query', '-Cv')
        self.log.info(status)

        self.log.info('-----获取当前首备节点的节点ID-----')
        p_status = status.splitlines()[-1].split('|')
        for i in p_status:
            if 'S Primary' in i or 'Promoting' in i:
                self.current_p_id = int(i.split()[0])
                self.log.info(f'---{self.current_p_id}---节点升为首备节点')

        text = '-----step6:在新的首备节点内查看表数据  expect:查看成功-----'
        self.log.info(text)

        self.log.info('---等待新的首备节点数据同步---')
        flag = self.s_comsh[int(self.current_p_id-1)].check_data_consistency()
        self.assertTrue(flag, '数据同步失败')

        p_select = self.s_comsh[int(self.current_p_id-1)].execut_db_sql(sql2)
        self.log.info(p_select)
        self.assertEqual(p_select.splitlines()[-2].strip(), select_res.splitlines()[-2].strip(), '查询结果不一致')

        text = '-----step7:在新的从备节点查看表数据  expect:查看失败,从备节点不可读-----'
        self.log.info(text)

        self.log.info('---获取当前从备节点的节点ID-----')
        for i in p_status:
            if 'S Standby' in i:
                self.current_p_id = int(i.split()[0])
                self.log.info(f'---{self.current_p_id}---节点为从备节点')

        ss_select = self.s_comsh[int(self.current_p_id-1)].execut_db_sql(sql2)
        self.log.info(ss_select)
        self.assertIn(self.constant.disaster_read_fail_msg, ss_select, '从备节点查询成功')

    def tearDown(self):
        text = '-----step8:清理环境，恢复环境初始状态  expect:清理成功，恢复环境成功-----'
        self.log.info(text)

        self.log.info('---恢复环境节点状态---')
        start = self.s_comsh[0].exec_cm_ctl('start')
        self.log.info(start)

        drop = f"drop table {self.table_new} cascade;"
        self.log.info(drop)
        drop_res = self.p_pri_sh.execut_db_sql(drop)
        self.log.info(drop_res)

        for param in self.param_list[:2]:
            reset_param = self.s_comsh[0].execute_gsguc('set',
                                                      self.constant.GSGUC_SUCCESS_MSG,
                                                      f'{param} = {self.default_value[self.param_list.index(param)]}')
            self.log.info(reset_param)

        restart = self.s_comsh[0].restart_db_cluster(False)
        self.log.info(restart)

        for param in self.param_list[:2]:
            reset_value = self.s_comsh[0].execut_db_sql(f'show {param};')
            self.log.info(reset_value)
            self.assertEqual(reset_value.splitlines()[-2].strip(),
                             self.default_value[self.param_list.index(param)], '参数恢复失败')

        recovery = self.s_comsh[0].exec_cm_ctl(mode='switchover', param='-a')
        self.log.info(recovery)

        status = self.s_comsh[0].exec_cm_ctl('query', '-Cv')
        self.log.info(status)
        self.assertIn('cluster_state  : Normal', status, '集群状态异常')
        self.assertIn(self.constant.cm_switchover_success_msg, recovery, '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, drop_res, '删除表失败')
        self.log.info(f'----{os.path.basename(__file__)} end----')
